package kafka;

import org.apache.kafka.clients.admin.KafkaAdminClient;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.annotation.PartitionOffset;
import org.springframework.kafka.annotation.TopicPartition;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.Acknowledgment;

import java.util.ArrayList;
import java.util.List;

public class KafkaTest {
    @Autowired
    private KafkaTemplate kafkaTemplate;
    @Autowired
    private JmsTemplate jmsTemplate;
    @Autowired
    private ConsumerFactory consumerFactory;
    @Autowired
    private KafkaAdminClient kafkaAdminClient;
    @Autowired
    private KafkaProducer kafkaProducer;
    @Autowired
    private KafkaConsumer kafkaConsumer;

    /**
     * 消费者
     */
    @KafkaListener(topics = {"da", "dwa"}, concurrency = "20", containerFactory = "batchFactory")
    public void consumer(String message, Acknowledgment ack) {
        ack.acknowledge();
    }

    /**
     * 消费者--指定主题，分区
     */
    @KafkaListener(id = "pp", autoStartup = "false",
            topicPartitions = @TopicPartition(topic = "topic1",
                    partitions = "0-5, 7, 10-15"))
    public void process(String in) {

    }

    /**
     * 消费者--指定主题，分区，初始的位移
     */
    @KafkaListener(id = "thing3", topicPartitions =
            {@TopicPartition(topic = "topic1", partitions = {"0", "1"},
                    partitionOffsets = @PartitionOffset(partition = "*", initialOffset = "0"))
            })
    public void listen(ConsumerRecord<?, ?> record) {
        //...消费，若未指定key，则为null
    }

    /**
     * 批量消费消息
     */
    @KafkaListener(topics = "teemo", id = "consumer", containerFactory = "batchFactory")
    public void listen(List<ConsumerRecord<?, ?>> list, Acknowledgment ack) {
        List<String> messages = new ArrayList<>();
        for (ConsumerRecord<?, ?> record : list) {
            // 获取消息
        }
        ack.acknowledge();
        if (messages.size() > 0) {
            // 更新索引
        }

    }

    /**
     * 生产者
     */
    public void provider() {
        kafkaTemplate.send("主题", 21, "密钥", "消息");
    }
}
